package com.hefei.garden.config.kafka.config;

import com.google.common.collect.Lists;
import com.google.common.collect.Maps;
import com.hefei.garden.config.kafka.annotation.KafkaInterceptorOrder;
import com.hefei.garden.config.kafka.container.KafkaOrderDefinition;
import com.hefei.garden.config.kafka.partitioner.CounterPartitioner;
import lombok.SneakyThrows;
import org.apache.kafka.clients.producer.ProducerConfig;
import org.apache.kafka.clients.producer.ProducerInterceptor;
import org.apache.kafka.common.serialization.StringSerializer;
import org.springframework.beans.factory.annotation.Value;
import org.springframework.beans.factory.config.BeanDefinition;
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.ClassPathScanningCandidateComponentProvider;
import org.springframework.context.annotation.Configuration;
import org.springframework.core.type.filter.AssignableTypeFilter;
import org.springframework.kafka.annotation.EnableKafka;
import org.springframework.kafka.core.DefaultKafkaProducerFactory;
import org.springframework.kafka.core.KafkaTemplate;
import org.springframework.util.CollectionUtils;

import java.util.*;

/**
 * @author: devil
 * @Date: 2022/8/21 16
 * @Description:
 */
@Configuration
@EnableKafka
public class KafkaProducerConfig {

//    @Value("${spring.kafka.servers}")
//    private String servers;
//    @Value("${spring.kafka.producer.retries}")
//    private int retries;
//    @Value("${spring.kafka.producer.batchSize}")
//    private int batchSize;
//    @Value("${spring.kafka.producer.linger}")
//    private int linger;
//    @Value("${spring.kafka.producer.bufferMemory}")
//    private int bufferMemory;
//    @Value("${spring.kafka.producer.acks}")
//    private String acks;
//    @Value("${spring.kafka.producer.maxRequestSize}")
//    private String maxRequestSize;
//    @Value("${spring.kafka.producer.compressionType}")
//    private String compressionType;
//    @Value("${spring.kafka.producer.connectionsMaxIdleMs}")
//    private String connectionsMaxIdleMs;
//    @Value("${spring.kafka.producer.receiveBufferBytes}")
//    private String receiveBufferBytes;
//    @Value("${spring.kafka.producer.sendBufferBytes}")
//    private String sendBufferBytes;
//    @Value("${spring.kafka.producer.requestTimeoutMs}")
//    private String requestTimeoutMs;
//
//    @Bean
//    public KafkaTemplate<String, String> kafkaTemplate() {
//        Map<String, Object> props = Maps.newHashMapWithExpectedSize(7);
//        props.put(ProducerConfig.BOOTSTRAP_SERVERS_CONFIG, servers);
//        props.put(ProducerConfig.RETRIES_CONFIG, retries);
//        props.put(ProducerConfig.BATCH_SIZE_CONFIG, batchSize);
//        props.put(ProducerConfig.LINGER_MS_CONFIG, linger);
//        props.put(ProducerConfig.BUFFER_MEMORY_CONFIG, bufferMemory);
//        props.put(ProducerConfig.KEY_SERIALIZER_CLASS_CONFIG, StringSerializer.class);
//        props.put(ProducerConfig.VALUE_SERIALIZER_CLASS_CONFIG, StringSerializer.class);
//        props.put(ProducerConfig.ACKS_CONFIG, acks);
//        props.put(ProducerConfig.MAX_REQUEST_SIZE_CONFIG, maxRequestSize);
//        props.put(ProducerConfig.COMPRESSION_TYPE_CONFIG, compressionType);
//        props.put(ProducerConfig.CONNECTIONS_MAX_IDLE_MS_CONFIG, connectionsMaxIdleMs);
//        props.put(ProducerConfig.RECEIVE_BUFFER_CONFIG, receiveBufferBytes);
//        props.put(ProducerConfig.SEND_BUFFER_CONFIG, sendBufferBytes);
//        props.put(ProducerConfig.REQUEST_TIMEOUT_MS_CONFIG, requestTimeoutMs);
//        // 拦截器
//        props.put(ProducerConfig.INTERCEPTOR_CLASSES_CONFIG, this.getAllInterceptor());
//        DefaultKafkaProducerFactory<String, String> factory = new DefaultKafkaProducerFactory<>(props);
//        return new KafkaTemplate<>(factory);
//    }
//
//    /**
//     * 获取所有实现的拦截器
//     */
//    @SneakyThrows
//    public String getAllInterceptor() {
//        String basePackage = "com.hefei.garden";
//        // spring提供的方法获取所有KafkaProducerInterceptor的子类
//        ClassPathScanningCandidateComponentProvider provider = new ClassPathScanningCandidateComponentProvider(false);
//        provider.addIncludeFilter(new AssignableTypeFilter(ProducerInterceptor.class));
//        Set<BeanDefinition> components = provider.findCandidateComponents(basePackage);
//        if (CollectionUtils.isEmpty(components)) {
//            return null;
//        }
//        List<KafkaOrderDefinition> interceptors = Lists.newArrayList();
//        for (BeanDefinition component : components) {
//            // 获取拦截器类名
//            String beanClassName = component.getBeanClassName();
//            Class<?> aClass = Class.forName(beanClassName);
//            // 获取类注解
//            KafkaInterceptorOrder interceptorAnnotation = aClass.getAnnotation(KafkaInterceptorOrder.class);
//            int order = interceptorAnnotation.order();
//            KafkaOrderDefinition kafkaOrderDefinition = new KafkaOrderDefinition();
//            kafkaOrderDefinition.setName(beanClassName);
//            kafkaOrderDefinition.setOrder(order);
//            // 便于最后的排序
//            interceptors.add(kafkaOrderDefinition);
//        }
//        // 根据order进行排序
//        interceptors.sort(Comparator.comparing(KafkaOrderDefinition::getOrder));
//        StringJoiner stringJoiner = new StringJoiner(",");
//        interceptors.forEach(kafkaOrderDefinition -> stringJoiner.add(kafkaOrderDefinition.getName()));
//        return stringJoiner.toString();
//    }

}
